package cn.tedu.charging.order.mqtt;

import cn.tedu.charging.common.pojo.JsonResult;
import cn.tedu.charging.common.pojo.param.ChargingProcessParam;
import cn.tedu.charging.common.pojo.vo.ChargingProcessVO;
import cn.tedu.charging.order.constant.Constants;
import cn.tedu.charging.order.constant.MqttConstant;
import cn.tedu.charging.order.constant.OrderStatusConstant;
import cn.tedu.charging.order.dao.repository.ChargingProcessRepository;
import cn.tedu.charging.order.feign.CostClient;
import cn.tedu.charging.order.pojo.dto.ChargingDTO;
import cn.tedu.charging.order.pojo.dto.ChargingProcessDTO;
import cn.tedu.charging.order.pojo.dto.ChargingResultDto;
import cn.tedu.charging.order.pojo.po.ChargingBillFailPO;
import cn.tedu.charging.order.pojo.po.ChargingBillSuccessPO;
import cn.tedu.charging.order.pojo.po.ChargingProcessPO;
import cn.tedu.charging.utils.JsonUtils;
import com.alibaba.nacos.shaded.org.checkerframework.checker.units.qual.A;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.weaver.ast.Or;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import sun.rmi.runtime.Log;

/**
 * EMQX的消息中间的 消费者
 * 订单服务 用来 接收(订阅 subscribe) 设备发送的消息
 * 1.订阅topic
 *  4个方法里，哪个放个来订阅topic 比较合理 连接成功 connectComplete 订阅 topic
 *
 * 2.接收消息
 */
@Slf4j
public class MqttConsumer implements MqttCallbackExtended {

    private MqttClient mqttClient;

    private MqttContext mqttContext;

    // 进行方法重构
    public MqttConsumer(MqttClient mqttClient,MqttContext mqttContext) {
        this.mqttClient = mqttClient;
        this.mqttContext = mqttContext;
    }

    // 连接成功
    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.debug("MqttConsumer#connectComplete方法被调了，表示连接成功,"+"reconnect:{},serverURI:{}",reconnect,serverURI);
        try {
            log.debug("连接成功，订阅开始充电指令的响应topic:{}", MqttConstant.TOPIC_CHARGING_RESULT);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_RESULT);   // 这是一个订阅者(消费者subscribe)，订阅了 topic；MQTT.fx（模拟设备publish）是一个发布者，用来测试和模拟设备发送消息
            log.debug("连接成功，订阅开始充电指令的响应topic:{}，成功", MqttConstant.TOPIC_CHARGING_RESULT);

            /**
             * 同步充电进度
             */
            log.debug("连接成功，用来接收设备同步的充电状态，调用topic:{}",MqttConstant.TOPIC_CHARGING_PROCESS);
            mqttClient.subscribe(MqttConstant.TOPIC_CHARGING_PROCESS);  // 这是一个订阅者(消费者subscribe)，订阅了 topic；MQTT.fx（模拟设备publish）是一个发布者，用来测试和模拟设备发送消息
            log.debug("连接成功，用来接收设备同步的充电状态，调用topic:{}，成功",MqttConstant.TOPIC_CHARGING_PROCESS);

        } catch (MqttException e) {
            throw new RuntimeException(e);
        }
    }

    // 连接丢失
    @Override
    public void connectionLost(Throwable throwable) {
        log.debug("MqttConsumer#connectionLost方法被调用了，表示连接丢失",throwable);
    }

    // 接收消息

    /**
     * 开始充电指令 订单服务 --> emqx --> 设备
     * 开始充电响应 设备 --> emqx(topic/start/result) --> 订单服务
     * 同步充电进度 设备 --> emqx(topic/charging/progress) --> 订单服务
     * messageArrived 指的是 设备发送的消息，到达了订单服务
     * @param topic
     * @param message
     * @throws Exception
     */
    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {    // EMQX 推送消息到 订单服务
        log.debug("MqttConsumer#messageArrived方法被调用了，表示消息到达,"+"topic:{},message:{}",topic,message);

        /**
         * 有多个topic需要处理
         */
        if (MqttConstant.TOPIC_CHARGING_RESULT.equals(topic)){
            log.debug("处理开始充电结果消息");
            handleChargingResult(message);    // 订单服务 调用 handleChargingProcess 方法
        }else if (MqttConstant.TOPIC_CHARGING_PROCESS.equals(topic)){
            log.debug("处理同步充电进度消息");
            handleChargingProcess(message);
        }
    }

    /**
     * 处理同步充电进度消息  设备同步充电的状态
     * 这部分是 都是 订单服务处理
     * @param message
     */
    /*@Autowired
    private CostClient costClient;
    不能使用这种方法，因为MqttConsumer 不在容器里面，而@Autowired在容器里面
    */
    private void handleChargingProcess(MqttMessage message) {   // message 是设备发布出来的消息
        log.debug("处理充电进度:{}",message);   // message 是设备发布出来的消息 "gunId": 1
        // 这里定义一个全局变量，不然try之后的无法使用
        ChargingProcessDTO chargingProcessDTO = null;
        try {
            chargingProcessDTO = JsonUtils.fromJson(message.toString(), ChargingProcessDTO.class);
        } catch (Exception e) {
            log.error("设备发送的充电进度消息格式{},有问题,请检查!",message,e);
        }
        if (chargingProcessDTO != null){
            // 计算电费
            ChargingProcessVO orderCost = getOrderCost(chargingProcessDTO);
            // 即通过 CostClient[订单服务]（OpenFeign）调用远程的计价服务(通过的就是 Feign HTTP 的调用)
            log.debug("计价返回的结果:{}",orderCost);    // 返回价格结果
            if (orderCost != null){
                // 保存充电进度数据
                saveChargingProcess(chargingProcessDTO,orderCost);
                // 推送充电数据给用户
                sendChargingProcess2User();
                // 判断余额 先充值，后充电
                checkUserBalance();
                // 判断设备的状态(温度...)
                checkDeviceStatus();
                // 判断是否充满
                checkOrderStatus();
            }else {
                log.error("获取价格失败:{},打印错误日志,接入告警系统,告警告警",chargingProcessDTO);
            }
        }
    }

    /**
     * 计算电费
     */
    private ChargingProcessVO getOrderCost(ChargingProcessDTO chargingProcessDTO){
        ChargingProcessParam chargingProcessParam = new ChargingProcessParam();
//        chargingProcessParam.setGunId(chargingProcessDTO.getGunId());
//        chargingProcessParam.setStationId(chargingProcessDTO.getStationId());
        BeanUtils.copyProperties(chargingProcessDTO,chargingProcessParam);
        log.debug("调用计价服务入参:{}",chargingProcessParam);
        CostClient costClient = mqttContext.getCostClient();
        JsonResult<ChargingProcessVO> chargingProcessVOJsonResult = costClient.calculateCost(chargingProcessParam);
        log.debug("调用计价服务入参:{},出参:{}",chargingProcessParam,chargingProcessVOJsonResult);
        if (chargingProcessVOJsonResult != null){
            return chargingProcessVOJsonResult.getData();
        }
        return null;
    }

    /*@Autowired    无法注入，使用MqttContext进行注入
    ChargingProcessRepository chargingProcessRepository;*/
    /**
     * 使用ES进行保存充电数据
     * @param chargingProcessDTO
     */
    private void saveChargingProcess(ChargingProcessDTO chargingProcessDTO,ChargingProcessVO chargingProcessVO) {
        log.debug("保存充电数据:{}",chargingProcessDTO);
        ChargingProcessPO chargingProcessPO = new ChargingProcessPO();
        chargingProcessPO.setId(System.currentTimeMillis());    // 把时间当做id
        BeanUtils.copyProperties(chargingProcessDTO,chargingProcessPO);   // 数据
        chargingProcessPO.setTotalCost(chargingProcessVO.getTotalCost());  // 获取总金额
        mqttContext.getChargingProcessRepository().save(chargingProcessPO);   // 保存到 ES 数据中
        log.debug("保存充电数据:{},成功",chargingProcessDTO);
    }


    private void checkOrderStatus() {
    }

    private void checkDeviceStatus() {
        
    }

    private void checkUserBalance() {
    }

    private void sendChargingProcess2User() {
    }




    /**
     * 处理开始充电结果消息，设备同步的开始充电结果 是否能够开始 开始充电成功  开始充电失败
     * @param message
     */
    private void handleChargingResult(MqttMessage message) {
        log.debug("充电进度：{}", message);
        // 订单服务接收到设备的响应后，处理消息
        // 把设备发送的消息Json转换为java对象 ChargingResultDto
        ChargingResultDto chargingResultDto = null;
        try {
            chargingResultDto = JsonUtils.fromJson(message.toString(), ChargingResultDto.class);
        } catch (Exception e) {
            log.error("设备发送的开始充电结果消息格式{},有问题,请检查!",message,e);
        }
        // 如果有响应，且结果不为空
        if (chargingResultDto != null){
            String result = chargingResultDto.getResult();
            // 如果充电成功
            if (Constants.RESULT_START_CHARGING_SUCCESS.equals(result)){
                saveSuccessOrder(chargingResultDto);   // 这里使用了 Refactor 中的 Extract Method 进行了方法重构，只是让代码看的更舒服
            }else {
                log.debug("通知充电用户，表示歉意，提醒用户更换充电枪");
                log.debug("通知设备管理人员，进行设备检修");
                saveFailOrder(chargingResultDto);    // 这里使用了 Refactor 中的 Extract Method 进行了方法重构，只是让代码看的更舒服

            }
        }
    }

    private void saveFailOrder(ChargingResultDto chargingResultDto) {
        log.debug("设备有响应，充电桩开始充电失败");
        ChargingBillFailPO chargingBillFailPO = new ChargingBillFailPO();
        chargingBillFailPO.setBillId(chargingResultDto.getOrderNo());
        chargingBillFailPO.setFailDesc("充电站故障-无法正常充电");
        log.debug("保存失败订单记录：{}",chargingBillFailPO);
        Integer row = mqttContext.getOrderFailMapper().saveOrderFailByMybatis(chargingBillFailPO);   // 输出影响的行数
        log.debug("保存失败订单记录：{},影响的行数：{},完成",chargingBillFailPO,row);
    }

    private void saveSuccessOrder(ChargingResultDto chargingResultDto) {
        log.debug("设备有响应，充电桩开始充电成功");
        ChargingBillSuccessPO chargingBillSuccessPO = new ChargingBillSuccessPO();
        chargingBillSuccessPO.setBillId(chargingResultDto.getOrderNo());   // 获取订单号
        // 设置订单状态为充电中
        chargingBillSuccessPO.setBillStatus(OrderStatusConstant.ORDER_STATUS_PROGRESS);
        log.debug("保存成功订单记录:{}",chargingBillSuccessPO);
        // 此时 MqttConsumer 不在容器(configuration)中，而我现在要用 Mapper 来保存（相当于：我现在要在不在容器中 MqttConsumer ，去使用在容器中的 Mapper） 怎么解决，如何关联？
        /**
         * 目前需要 成功的mapper，失败的mapper，甚至以后还需要其他的mapper
         * 方案一：修改构造器 MqttConsumer，每次增加依赖，都修改构造器 (在第28行的括号中，加入新的参数)
         * 方案二：再创建一个 MqttContext 类，加入 @Component 注解，放入到容器中
         */
        // 使用MyBatis-Plus BaseMapper 接口提供的通用插入方法
//                mqttContext.getOrderSuccessMapper().insert(chargingBillSuccessPO);

        // 使用 XML 方式定义的插入方法，如果想使用上面的 BaseMapper 插入方法，则需要把xml文件中sql语句注释掉
        mqttContext.getOrderSuccessMapper().saveOrderByMybatis(chargingBillSuccessPO);
        log.debug("保存成功订单记录:{},完成",chargingBillSuccessPO);
    }

    /**
     * 加锁  保证线程按照顺序执行 T2 先执行，T1 再执行，线程之间有了先后的顺序 排队
     *    T1  T2  T3 多个线程来抢锁
     *    T2 先成功抢到，T1 T3 需要等T2线程释放锁，T1 T3继续等待或者放弃
     *    T2:执行操作
     *    T2 OrderDTO orderDto = repository.getOrderByOrderNo(orderNo);
     *    T2 if(orderDto == null){    // 操作中
     *        T2 repository.saveOrder(orderDto);
     *    }
     *    操作结束：释放锁 T2
     *
     *    T1 获取到锁
     *    T1 OrderDTO orderDto = repository.getOrderByOrderNo(orderNo);
     *    T1 if(orderDto == null){    // 操作中  上一个线程T2已经保存数据成功了，T1 判断 orderDTO == null 不为真
     *        T1 repository.saveOrder(orderDto);    不执行
     *    }
     *    T1 释放锁
     *
     *    如果给 private synchronized void saveSuccessOrder(ChargingResultDto chargingResultDto) 加锁
     *    也只是加到本地锁，它只能互斥当前jvm的线程，不能互斥其他jvm的线程
     *
     *    在分布式下 要加分布式锁 通过Redis实现分布式锁的机制  Todo
     */




    // 消息发送完成
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("MqttConsumer#deliveryComplete方法被调了，表示消息发送完成"+"token:{}",token);
    }
}